package cn.bud.msgbus.consumer.server;

import cn.bud.msgbus.server.bean.MsgBody;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

/**
 * MsgBusDataQueue
 * 数据暂时存在cache中，
 * 后期需要考虑缓存数据持久化，防止未推送数据丢失的情况发生
 *
 * @author hxj
 * @version 1.0
 * @description
 * @date 2024-06-20 9:21
 */
@Component
@Slf4j
public class MsgBusDataQueue implements MsgBusDataPusher {

    private BlockingDeque<MsgBody> queue = new LinkedBlockingDeque<>(10000);
    @Override
    public void pushMsg(MsgBody msg) {
        try {
            boolean b = queue.offer(msg, 10, TimeUnit.SECONDS);
            if (!b) {
                // 删除1000个最前面的数据
                log.info("delete 1000 data, queue.size={}", queue.size());
                int rmCount = 0;
                for (int i = 0; i < 1000; i++) {
                    if (!queue.isEmpty()) {
                        rmCount++;
                        queue.poll(1, TimeUnit.MILLISECONDS);
                    } else {
                        break;
                    }
                }
                log.info("delete 1000 data, 实际删除数据{}个, queue.size={}", rmCount, queue.size());
            }
        } catch (Exception e) {
            log.info("error:", e);
        }
    }

    public MsgBody poll() {
        MsgBody msg = null;
        synchronized (queue) {
            try {
                msg = queue.poll(3, TimeUnit.SECONDS);
            }catch (Exception e) {
            }
        }
        return msg;
    }

    /**
     * 关闭前的操作
     * 1.关闭数据存取接口
     * 2.未推送数据持久化
     */
    public void stop() {
    }
}
